package com.dukk.espark.utils;

import org.apache.commons.lang.StringUtils;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;

/**
 * 自定义转换类
 * @version 2020-08-24
 * @author shengshi_feiyang@yeah.net
 */
public class Row2ObjUtil<T> implements Serializable {
    private static final long serialVersionUID = 5774467935765466231L;

    private static final Logger logger = LoggerFactory.getLogger(Row2ObjUtil.class);

    T t;

    /**
     * 现在我得转义功能还比较单一，只能完成简单数据类型得自动转义。
     * 有需要得自己扩充奥
     *
     * 支持得数据类型：long int string double
     */
    public T transform(T obj, Row row){

        StructType structType = row.schema();
        StructField structFields[] = structType.fields();
        Map<String, String> sparkFiledTypeMap = new HashMap<>();
        Map<String, String> sparkFiledUpNameMapping = new HashMap<>();
        for(StructField structField : structFields){
            String fieldName = structField.name();
            String typeName = structField.dataType().typeName();
            sparkFiledUpNameMapping.put(fieldName.toUpperCase(), fieldName);
            if(typeName.contains("decimal")){
                sparkFiledTypeMap.put(fieldName.toUpperCase(), "decimal".toUpperCase());
            }else {
                sparkFiledTypeMap.put(fieldName.toUpperCase(), typeName.toUpperCase());
            }

        }

        Class class_ = obj.getClass();
        Method[] methods = class_.getMethods();
        Map<String, Method> methodMap = new HashMap<>();
        for(Method method : methods){
            String methodName = method.getName();
            if(methodName.startsWith("set")){
                methodMap.put(methodName.replaceFirst("set", "").toUpperCase(), method);
            }
        }
        Field fields[] = class_.getDeclaredFields();
        for(Field field : fields){

            String fieldName = field.getName().toUpperCase();
            String typeName = field.getType().getTypeName().toUpperCase();

            if(fieldName.equals("SERIALVERSIONUID")){
                continue;
            }

            //String 类型得字段默认给赋值成 "" 不赋值
            if(typeName.contains("STRING")){
                try {
                    methodMap.get(fieldName).invoke(obj, "");
                } catch (Exception e) {
                    logger.error("##########字段转义失败#########类{},字段{},类型{},错误{}", class_.getName(),fieldName,typeName,e);
                    e.printStackTrace();
                }
            }

            if(null == sparkFiledTypeMap.get(fieldName) && StringUtils.isBlank(sparkFiledTypeMap.get(fieldName))){
                continue;
            }

            String sparkType = sparkFiledTypeMap.get(fieldName);

            String sparkFieldName = sparkFiledUpNameMapping.get(fieldName);

            if(StringUtils.isNotBlank(sparkType) && null != row.getAs(sparkFieldName)){
                //long类型赋值
                if(typeName.contains("LONG") && (sparkType.contains("LONG")  || sparkType.contains("DECIMAL") ||  sparkType.contains("STRING"))){
                    if(sparkType.contains("LONG")){
                        try {
                            methodMap.get(fieldName).invoke(obj,(Long)row.getAs(sparkFieldName));
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }
                    if(sparkType.contains("DECIMAL")){
                        try {
                            methodMap.get(fieldName).invoke(obj, ((BigDecimal)row.getAs(sparkFieldName)).longValue());
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }

                    if(sparkType.contains("STRING")){
                        try {
                            methodMap.get(fieldName).invoke(obj, (Long.parseLong(row.getAs(sparkFieldName))));
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }

                }
                //string 赋值
                if(typeName.contains("STRING") && sparkType.contains("STRING")){
                    try {
                        String strVal = (String)row.getAs(sparkFieldName);
                        if(StringUtils.isNotBlank(strVal)){
                            methodMap.get(fieldName).invoke(obj, strVal);
                        }
                    } catch (Exception e) {
                        logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                        e.printStackTrace();
                    }
                }
                //int 赋值
                if("INT".equals(typeName) && (sparkType.contains("INT")  || sparkType.contains("DECIMAL") ||  sparkType.contains("STRING"))){
                    if(sparkType.contains("INT")){
                        try {
                            methodMap.get(fieldName).invoke(obj,(int)row.getAs(sparkFieldName));
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }
                    if(sparkType.contains("DECIMAL")){
                        try {
                            methodMap.get(fieldName).invoke(obj, ((BigDecimal)row.getAs(sparkFieldName)).intValue());
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }

                    if(sparkType.contains("STRING")){
                        try {
                            methodMap.get(fieldName).invoke(obj, Integer.parseInt((String)row.getAs(sparkFieldName)));
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }
                }

                //double 赋值
                if("DOUBLE".equals(typeName) && (sparkType.contains("DOUBLE")  || sparkType.contains("DECIMAL") ||  sparkType.contains("STRING"))){
                    if(sparkType.contains("DOUBLE")){
                        try {
                            methodMap.get(fieldName).invoke(obj,(double)row.getAs(sparkFieldName));
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }
                    if(sparkType.contains("DECIMAL")){
                        try {
                            methodMap.get(fieldName).invoke(obj, ((BigDecimal)row.getAs(sparkFieldName)).doubleValue());
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }

                    if(sparkType.contains("STRING")){
                        try {
                            methodMap.get(fieldName).invoke(obj, Double.parseDouble((String)row.getAs(sparkFieldName)));
                        } catch (Exception e) {
                            logger.error("##########字段转义失败#########类{},字段{},类型{},spark类型{},错误{}", class_.getName(),fieldName,typeName,sparkType, e);
                            e.printStackTrace();
                        }
                    }
                }
            }
        }

        return obj;
    }



}
